package io.kiki.sba.registry.client.event;

import io.kiki.sba.registry.client.api.ClientConfig;
import io.kiki.sba.registry.client.api.EventBus;
import io.kiki.sba.registry.client.api.EventSubscriber;
import io.kiki.sba.registry.client.api.model.Event;
import io.kiki.sba.registry.client.factory.NamedThreadFactory;
import lombok.Getter;
import lombok.Setter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.*;

@Setter
@Getter
public class EventBusImpl implements EventBus {

    private static final Logger logger = LoggerFactory.getLogger(EventBusImpl.class);

    private ClientConfig clientConfig;

    private ConcurrentMap<Class<? extends Event>, CopyOnWriteArraySet<EventSubscriber>> eventClassToEventSubscriberSetMap = new ConcurrentHashMap<Class<? extends Event>, CopyOnWriteArraySet<EventSubscriber>>();

    private Executor executor;


    public EventBusImpl(ClientConfig clientConfig) {
        this.clientConfig = clientConfig;
        this.executor = new ThreadPoolExecutor(clientConfig.getObserverThreadCoreSize(), clientConfig.getObserverThreadMaxSize(), 0, TimeUnit.SECONDS, new LinkedBlockingDeque<Runnable>(clientConfig.getObserverThreadQueueLength()), new NamedThreadFactory("DefaultEventBusThread"));
    }


    @Override
    public void register(Class<? extends Event> eventClass, EventSubscriber eventSubscriber) {
        logger.debug("Register subscriber: {} of event: {}.", eventSubscriber, eventClass);
        eventClassToEventSubscriberSetMap.putIfAbsent(eventClass, new CopyOnWriteArraySet<>());
        CopyOnWriteArraySet<EventSubscriber> set = eventClassToEventSubscriberSetMap.get(eventClass);
        set.add(eventSubscriber);
    }


    @Override
    public void unRegister(Class<? extends Event> eventClass, EventSubscriber eventSubscriber) {
        CopyOnWriteArraySet<EventSubscriber> set = eventClassToEventSubscriberSetMap.get(eventClass);
        if (set != null) {
            set.remove(eventSubscriber);
            logger.debug("UnRegister subscriber: {} of event: {}.", eventSubscriber, eventClass);
        }
    }

    @Override
    public void post(final Event event) {
        if (!isEnable()) {
            return;
        }
        CopyOnWriteArraySet<EventSubscriber> set = eventClassToEventSubscriberSetMap.get(event.getClass());
        if (null != set && !set.isEmpty()) {
            for (final EventSubscriber eventSubscriber : set) {
                if (eventSubscriber.isSync()) {
                    handleEvent(eventSubscriber, event);
                } else { // 异步
                    executor.execute(new Runnable() {
                        @Override
                        public void run() {
                            handleEvent(eventSubscriber, event);
                        }
                    });
                }
            }
        }
    }


    @Override
    public boolean isEnable() {
        return null != clientConfig && clientConfig.isEventBusEnable();
    }

    @Override
    public boolean isEnable(Class<? extends Event> eventClass) {
        CopyOnWriteArraySet<EventSubscriber> set = eventClassToEventSubscriberSetMap.get(eventClass);
        return isEnable() && set != null && !set.isEmpty();
    }

    private void handleEvent(final EventSubscriber eventSubscriber, final Event event) {
        try {
            eventSubscriber.onEvent(event);
        } catch (Throwable e) {
            logger.warn("Handle {} error", event.getClass(), e);
        }
    }
}
